package io.kiki.sba.registry.server.clientInterface.strategy.impl;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.core.constants.EventTypeConstants;
import io.kiki.sba.registry.core.model.PublisherRegister;
import io.kiki.sba.registry.core.model.RegisterResponse;
import io.kiki.sba.registry.server.clientInterface.converter.PublisherConverter;
import io.kiki.sba.registry.server.clientInterface.registry.Registry;
import io.kiki.sba.registry.server.clientInterface.strategy.PublisherHandlerStrategy;
import io.kiki.sba.registry.server.shared.util.DatumUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import static io.kiki.sba.registry.common.model.constants.ValueConstants.DEFAULT_INSTANCE_ID;


public class DefaultPublisherHandlerStrategy implements PublisherHandlerStrategy {
    private static final Logger logger = LoggerFactory.getLogger(DefaultPublisherHandlerStrategy.class);
    @Autowired
    protected Registry registry;


    @Override
    public void handlePublisherRegister(Channel channel, PublisherRegister publisherRegister, RegisterResponse registerResponse) {
        Publisher publisher = null;

        try {
            String ip = channel.getRemoteAddress().getAddress().getHostAddress();
            int port = channel.getRemoteAddress().getPort();
            publisherRegister.setIp(ip);
            publisherRegister.setPort(port);

            if (StringUtils.isBlank(publisherRegister.getZone())) {
                publisherRegister.setZone(ValueConstants.DEFAULT_ZONE);
            }

            if (StringUtils.isBlank(publisherRegister.getInstanceId())) {
                publisherRegister.setInstanceId(DEFAULT_INSTANCE_ID);
            }

            publisher = PublisherConverter.convert(publisherRegister);
            publisher.setProcessId(ip + ":" + port);

            handle(publisher, channel, publisherRegister, registerResponse);
        } catch (Throwable e) {
            handleError(publisherRegister, publisher, registerResponse, e);
        }
    }

    protected void handle(Publisher publisher, Channel channel, PublisherRegister publisherRegister, RegisterResponse registerResponse) {

        publisher.setClientUrl(new Url(channel.getRemoteAddress()));
        publisher.setServerUrl(new Url(channel.getLocalAddress()));
        final String eventType = publisherRegister.getEventType();
        if (EventTypeConstants.REGISTER.equals(eventType)) {
            registry.register(channel, publisher);
        }
        if (EventTypeConstants.UNREGISTER.equals(eventType)) {
            registry.unRegister(publisher);
        }

        registerResponse.setSuccess(true);
        registerResponse.setVersion(publisher.getVersion());
        registerResponse.setRegistId(publisherRegister.getRegistId());
        registerResponse.setMessage("Publisher register success!");
        log(true, publisherRegister, publisher);
    }

    private void log(boolean success, PublisherRegister publisherRegister, Publisher publisher) {
        // [Y|N],[R|U|N],app,zone,dataInfoId,registerId,version,registerTimestamp,clientVersion,clientIp,clientPort,dataSize
        long size = DatumUtils.getDataListSize(publisherRegister.getDataList());
        Metrics.Access.pubCount(success);
        Metrics.Access.pubSize("registry", publisherRegister.getGroup(), size);
        logger.info("{},{},{},{},{},G={},I={},{},{},{},{},{},{},size={},pb={},attrs={}", success ? 'Y' : 'N', EventTypeConstants.getEventTypeFlag(publisherRegister.getEventType()), publisherRegister.getAppName(), publisherRegister.getZone(), publisher == null ? publisherRegister.getDataId() : publisher.getDataId(), publisherRegister.getGroup(), publisherRegister.getInstanceId(), publisherRegister.getRegistId(), publisherRegister.getVersion(), publisher == null ? "" : publisher.getRegisterTimestamp(), publisher == null ? "" : publisher.getClientVersion(), publisherRegister.getIp(), publisherRegister.getPort(), size, publisher == null ? "0" : publisher.attributesSize());
    }

    protected void handleError(PublisherRegister publisherRegister, Publisher publisher, RegisterResponse registerResponse, Throwable e) {
        log(false, publisherRegister, publisher);
        RegisterLogs.logError(publisherRegister, "Publisher", registerResponse, e);
    }
}
